package com.example.aggregation;

import com.example.bean.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MapDemo
 * Package: com.example.transformation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-18
 * Time: 16:03
 */

public class ReduceDemo {
    public static void main(String[] args) throws Exception {
        //1.创建环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //fromElements 是直接赋值数据
        DataStreamSource<WaterSensor> data = env.fromElements(
                //赋值WaterSensor对象的数据
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        KeyedStream<WaterSensor, String> sensorKS = data.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                //按照ID进行分组
                return value.getId();
            }
        });
        //规约聚合 也是只能在keyBy之后使用
        //两两聚合 输入类型 必须 等于输出类型
        SingleOutputStreamOperator<WaterSensor> reduce = sensorKS.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor val1, WaterSensor val2) throws Exception {
                //能进来的数据 id都是一样的
                System.out.println("val1: " + val1);
                System.out.println("val2: " + val2);
                return new WaterSensor(val1.getId(), val2.getTs(), val1.getVc() + val2.getVc());

            }
        });
        reduce.print();
        env.execute();
    }

}
